package com.test.cn.kafka

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.test.cn.kafka.message.{HeartBeat, RegisterWork, RegisteredWorker, SendHeartBeat}
import com.typesafe.config.ConfigFactory

class WorkerClient(masterHost:String,masterPort:Int,cores:Int,memory:Int) extends Actor {

  //保存使用akka-url获得的client,该对象可以发生消息
  var client:ActorSelection=_
  //worker唯一标识
  val id_client: String = UUID.randomUUID().toString
  override def preStart(): Unit = {
    println("开始进行注册")

    client = context.actorSelection(s"akka.tcp://masterSystem@${masterHost}:${masterPort}/user/masterActor")

    client ! RegisterWork(id_client,memory,cores)
  }

  override def receive: Receive = {
    case RegisteredWorker(namePort)=>{
      println(s"注册成功 master为$namePort ")
      import scala.concurrent.duration._
      import context.dispatcher
      context.system.scheduler.schedule( 0 millis, 5000 millis, self, SendHeartBeat)
    }
    case SendHeartBeat=>{
      client ! HeartBeat(id_client)
    }


  }
}



object  WorkerClient extends App {

    val masterHost = "127.0.0.1"
    var masterPort = 9999
    var cores = 64
    var memory = 1024


    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = ${masterHost}
         |akka.remote.netty.tcp.port=8887
      """.stripMargin


    val conf = ConfigFactory.parseString(str)

    val actorSystem = ActorSystem("clientServer",conf)

    actorSystem.actorOf(Props(new WorkerClient(masterHost,masterPort,cores,memory)),"clientServer")


}